package cn.doitedu.rtmk.demo8;

import cn.doitedu.rtmk.beans.RuleMetaBean;
import cn.doitedu.rtmk.beans.UserEvent;
import cn.doitedu.rtmk.interfaces.IRuleCalculatorV3;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.RoaringBitmap;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public class RuleEngineCoreFunction8 extends KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject> {

    Map<Integer, IRuleCalculatorV3> ruleCalculatorMap = new HashMap<>();


    @Override
    public void open(Configuration parameters) throws Exception {
    }

    /**
     * 处理主流数据的方法
     * 静态圈选条件：月平均消费额>200的用户
     * 触发条件：当他发生A行为时（event_id='page_load', properties['url'] = '/page/001'），命中规则
     */
    @Override
    public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        // 遍历运算机集合中的每一个规则运算机
        for (Map.Entry<Integer, IRuleCalculatorV3> entry : ruleCalculatorMap.entrySet()) {

            IRuleCalculatorV3 ruleCalculator = entry.getValue();
            // 传入当前的用户行为，交给规则运算机去处理
            ruleCalculator.calc(userEvent, out);

        }

    }

    /**
     * 处理广播流数据的方法
     */
    @Override
    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        // 取出新收到的规则元信息中的： 规则id 和 预圈选人群bytes
        int rule_id = (int) ruleMetaBean.getRule_id();

        /**
         * 此处是规则下线、停用功能处理
         * rowKind ==>  0: +I  , 1:-U  ,2: +U  , 3: -D
         */
        int rowKind = ruleMetaBean.getRowKind();
        // 规则元数据表中的 规则删除和 规则停用，在我们后端的规则引擎中，都代表将运算机移除
        if(rowKind == 3 || ruleMetaBean.getOnline_status() ==  0){
            ruleCalculatorMap.remove(rule_id);
            log.warn("移除了一个规则:{}",rule_id);
        }else {

            byte[] pre_select_users = ruleMetaBean.getPre_select_users();

            int rule_model_id = ruleMetaBean.getRule_model_id();
            String rule_param_json = ruleMetaBean.getRule_param_json();

            String rule_model_calc_groovy_code = ruleMetaBean.getRule_model_calc_groovy_code();


            // 反序列化出预圈选人群的bitmap对象
            RoaringBitmap preSelectUsersBitmap = RoaringBitmap.bitmapOf();
            ByteBuffer byteBuffer = ByteBuffer.wrap(pre_select_users);
            preSelectUsersBitmap.deserialize(byteBuffer);

            log.warn("收到一条新规则的元信息, 规则id:{},所属模型id:{},规则预圈选人群人数:{},规则参数:{}", rule_id, rule_model_id, preSelectUsersBitmap.getCardinality(), rule_param_json);


            /**
             * 利用groovy编译器，编译收到的规则所对应的规则运算机的代码
             * 并反射出实例对象
             * 然后传入预圈选人群、规则实例参数等信息来对规则运算机实例对象初始化
             */
            GroovyClassLoader groovyClassLoader = new GroovyClassLoader();
            Class aClass = groovyClassLoader.parseClass(rule_model_calc_groovy_code);
            IRuleCalculatorV3 calculator = (IRuleCalculatorV3) aClass.newInstance();
            calculator.open(preSelectUsersBitmap, JSON.parseObject(rule_param_json), getRuntimeContext());

            // 将规则运算机放入运算机集合
            ruleCalculatorMap.put(rule_id, calculator);
        }
    }

}
